package com.源码分析.以编程方式创建流;

import java.util.ArrayList;
import java.util.List;

import org.junit.Test;

import reactor.core.publisher.Flux;

/**作为一个更高级的创建 Flux 的方式， create 方法的生成方式既可以是同步， 也可以是异步的，并且还可以每次发出多个元素。

该方法用到了 FluxSink，后者同样提供 next，error 和 complete 等方法。 与 generate 不同的是，create 不需要状态值，另一方面，它可以在回调中触发 多个事件（即使是在未来的某个时间）。

 	create 有个好处就是可以将现有的 API 转为响应式，比如监听器的异步方法。

此外，既然 create 可以是异步地，并且能够控制背压，你可以通过提供一个 OverflowStrategy 来定义背压行为。

IGNORE： 完全忽略下游背压请求，这可能会在下游队列积满的时候导致 IllegalStateException。

ERROR： 当下游跟不上节奏的时候发出一个 IllegalStateException 的错误信号。

DROP：当下游没有准备好接收新的元素的时候抛弃这个元素。

LATEST：让下游只得到上游最新的元素。

BUFFER：（默认的）缓存所有下游没有来得及处理的元素（这个不限大小的缓存可能导致 OutOfMemoryError）。

 	Mono 也有一个用于 create 的生成器（generator）—— MonoSink，它不能生成多个元素， 因此会抛弃第一个元素之后的所有元素。
 * @author DELL
 *
 */
// Create()方法跟Generate()比较来看，有这么几个不同：
//Create()可以同步、也可以异步。而Generate()是同步的。
//Create()可以每次发出多个元素，而Generate()每次发出一个。
//Create()不需要状态值。
//Create()可以在回调中触发多个事件。
//Create()是多线程的。
//并且， Create()最常用的场景是：将现有的API转化为响应式，比如监听器的异步方法。
// Create是Pull和Push模式的混合，意思是：
// 1.上游在数据准备完成后可以将其推送给下游。
// 2.下游可以主动从上游中拉取数据。
public class Create {

	// 假设你有一个监听器 API，它按 chunk 处理数据，有两种事件：（1）一个 chunk 数据准备好的事件；（2）处理结束的事件
	public static interface MyEventListener<T> {
	    void onDataChunk(List<T> chunk);
	    void processComplete();
	}

	public static class MyEventProcessor {

		private MyEventListener<String> myEventListener;

		public void register(MyEventListener<String> myEventListener) {
			this.myEventListener = myEventListener;
		}

		public void add(ArrayList<String> chunk) {
			myEventListener.onDataChunk(chunk);
		}

	}

	@Test
	public void test() {
		MyEventProcessor myEventProcessor = new MyEventProcessor();

		// 所有这些都是在 myEventProcessor 执行时异步执行的。
		Flux<String> bridge = Flux.create(sink -> {
			// 桥接 MyEventListener
			myEventProcessor.register(new MyEventListener<String>() {

				// 每一个 chunk 的数据转化为 Flux 中的一个元素
				@Override
				public void onDataChunk(List<String> chunk) {
					for (String s : chunk) {
						sink.next(s);
					}
				}

				// processComplete 事件转换为 onComplete。
				@Override
				public void processComplete() {
					sink.complete();
				}
			});
		});
		bridge.subscribe(System.out::println);

		myEventProcessor.add(new ArrayList<String>(){{add("aaa");add("bbb");add("ccc");}});
		myEventProcessor.add(new ArrayList<String>(){{add("ddd");add("eee");}});
	}

}
